package com.deng.eventflow.core.tmf;

import com.deng.eventflow.core.Pullable;
import com.deng.eventflow.core.SynchronousProcessor;
import com.deng.eventflow.core.exception.ProcessorException;
import com.deng.eventflow.core.exception.PullableException;

public abstract class Sink extends SynchronousProcessor {
    public Sink()
    {
        this(1);
    }

    public Sink(int in_arity)
    {
        super(in_arity, 0);
    }

    public final void pull()
    {
        Object[] inputs = new Object[getInputArity()];
        for (int i = 0; i < getInputArity(); i++)
        {
            Pullable p = m_inputPullables[i];
            inputs[i] = p.pullSoft();
        }
        try
        {
            compute(inputs, null);
        }
        catch (ProcessorException e)
        {
            throw new PullableException(e);
        }
    }

    /**
     * Tells the sink to pull events from the pipeline
     */
    public final void pullHard()
    {
        Object[] inputs = new Object[getInputArity()];
        for (int i = 0; i < getInputArity(); i++)
        {
            Pullable p = m_inputPullables[i];
            inputs[i] = p.pull();
        }
        try
        {
            compute(inputs, null);
        }
        catch (ProcessorException e)
        {
            throw new PullableException(e);
        }
    }
}
